package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.promo.cache.CacheManager;
import com.mall.promo.dal.persistence.PromoItemMapper;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.mapstruct.Mapping;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Component
public class PromoOrderProducer {


    private TransactionMQProducer producer;

    @Autowired
    PromoItemMapper itemMapper;


    @Autowired
    CacheManager cacheManager;

    @Autowired
    RedissonClient redissonClient;

    @PostConstruct
    public void init() {
        producer = new TransactionMQProducer("promo_order_producer");

        producer.setNamesrvAddr("127.0.0.1:9876");

        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                Map<String, Long> param = (Map<String, Long>) arg;
                Long productId = param.get("productId");
                Long psId = param.get("psId");

                // 执行本地事物，扣减秒杀商品库存
                Integer effectiveRow = 0;

                // 在库存扣减的过程中可能会存在超卖的情况【集群或者mysql对数量没有检查】
                // 引入第三方redis，来加锁保证每一次扣除库存的过程只能有一个人进行。
                //  redis分布式锁【Push_sub】底层是由发布订阅模型解决的。
                String promoStockLockKey = "promo_item_stock: " + psId + "-" + productId;
                Lock lock = redissonClient.getLock(promoStockLockKey);

                lock.lock();
                try {
                    effectiveRow = itemMapper.decreaseStock(productId, psId);
                } finally {
                    lock.unlock();
                }

                String localTransactionKey = "promo_local_transaction: " + msg.getTransactionId();
                if (effectiveRow < 1) {
                    // 保存本地事物执行的结果
                    cacheManager.setCache(localTransactionKey, "failed", 1);

                    // 秒杀商品已经售罄了，不需要让订单服务，去生成秒杀商品订单
                    return  LocalTransactionState.ROLLBACK_MESSAGE;
                }

                // 保存本地事物的执行结果
                cacheManager.setCache(localTransactionKey, "success", 1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                String localTransactionKey = "promo_local_transaction: " + msg.getTransactionId();

                String result = cacheManager.checkCache(localTransactionKey);
                if (result == null || result.isEmpty()) {
                    return LocalTransactionState.UNKNOW;
                }

                if ("success".equals(result)) {
                    // 本地事物执行成功
                    return LocalTransactionState.COMMIT_MESSAGE;
                }

                // 返回本地事物执行失败的结果
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });

        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }




    public boolean sendPromoOrderTransactionMessage(CreateSeckillOrderRequest request, Long productId, Long psId) {
        // 将参数对象转化为json字符串
        String data = JSON.toJSONString(request);

        // 准备事物消息
        Message message = new Message();
        message.setTopic("promo_order");
        message.setBody(data.getBytes(StandardCharsets.UTF_8));

        HashMap<String, Long> paramMap = new HashMap<>();
        paramMap.put("productId", productId);
        paramMap.put("psId", psId);


        TransactionSendResult sendResult = null;
        try {

            sendResult = producer.sendMessageInTransaction(message, paramMap);
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        if (sendResult != null) {
            LocalTransactionState myTransactionState = sendResult.getLocalTransactionState();
            if (LocalTransactionState.COMMIT_MESSAGE.equals(myTransactionState)) {
                // 本地事务执行成功，站在柔性事物最终一致性的角度，就可以认为用户秒杀下单成功过
                return true;
            }
        }

        //  要么发送事物消息未成功，要么本地事物执行失败
        return false;
    }
}
